package cn.com.bluemoon.bd.flink.runner;

import com.google.common.collect.Lists;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.List;

public abstract class AbstractProcessorBuilder {
    protected String[] args;
    protected List<AbstractProcessor> processorList = Lists.newArrayList();

    public AbstractProcessorBuilder(String[] args) {
        this.args = args;
    }

    public abstract AbstractProcessorBuilder build(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env);

    public void add(AbstractProcessor processor) {
        this.processorList.add(processor);
    }

    public void process() throws Exception {
        for (AbstractProcessor processor : processorList) {
            processor.process();
        }
    }
}
